package com.sh.data.engine.domain.datadev.flink.service.impl;

import com.sh.data.engine.common.util.FlinkUtil;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkCheckGrammarDomain;
import com.sh.data.engine.domain.datadev.flink.service.FlinkService;
import org.apache.calcite.sql.parser.SqlParseException;
import org.apache.calcite.sql.parser.SqlParserPos;
import org.springframework.stereotype.Service;

/**
 * @author: zigui.zdf
 * @description:
 * @date: 2021/6/11 13:56
 */
@Service
public class FlinkServiceImpl implements FlinkService {

    private static final String defaultEnv =
        "# job失败恢复策略\n# full（重启所有task）\n# region（重启可能引起任务失败的task）\njobmanager.execution.failover-strategy=region\n\n# JobManager运行时总内存大小\njobmanager.memory.process.size=768m\n\n# TaskManager运行时总内存大小\ntaskmanager.memory.process.size=1024m \n\n# 默认并行度\nparallelism.default=1\n\n# 每个TaskManager的并行度。一个slot对应一个core，默认值是1.一个并行度对应一个线程。总的内存大小要且分给不同的线程使用\ntaskmanager.numberOfTaskSlots=1";

    private static final String defaultFlinkxEnv =
        "# checkpoint保存时间间隔（单位：ms）\nflink.checkpoint.interval=60000\n\n# checkpoint 保留份数\nstate.checkpoints.num-retained=20";

    @Override
    public FlinkCheckGrammarDomain verifySql(String sqlStr) {
        FlinkCheckGrammarDomain domain = new FlinkCheckGrammarDomain();

        try {
            FlinkUtil.verifySql(sqlStr);
            domain.setCorrect(true);
        } catch (SqlParseException e) {
            final String message = e.getMessage();
            final SqlParserPos pos = e.getPos();

            domain.setCorrect(false);
            domain.setMsg(message);
            domain.setColumnNumber(pos.getColumnNum());
            domain.setLineNumber(pos.getLineNum());
            domain.setEndColumnNumber(pos.getEndColumnNum());
            domain.setEndLineNumber(pos.getEndLineNum());
        }
        return domain;
    }

    @Override
    public String getDefaultEnv() {
        return defaultEnv;
    }

    @Override
    public String getDefaultFlinkxEnv() {
        return defaultFlinkxEnv;
    }
}
